/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.iotdb.db.mpp.plan.execution.config.executor;

import com.google.common.util.concurrent.SettableFuture;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.consensus.ConfigNodeRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.executable.ExecutableManager;
import org.apache.iotdb.commons.executable.ExecutableResource;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFClassLoader;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.confignode.rpc.thrift.*;
import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.ConfigNodeInfo;
import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
import org.apache.iotdb.db.metadata.template.ClusterTemplateManager;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.mpp.plan.analyze.Analyzer;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.*;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowNodesInSchemaTemplateTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowPathSetTemplateTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowSchemaTemplateTask;
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeSinkTask;
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeTask;
import org.apache.iotdb.db.mpp.plan.statement.metadata.*;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.*;
import org.apache.iotdb.db.mpp.plan.statement.sys.sync.*;
import org.apache.iotdb.db.sync.SyncService;
import org.apache.iotdb.db.trigger.service.TriggerClassLoader;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.trigger.api.Trigger;
import org.apache.iotdb.trigger.api.enums.FailureStrategy;
import org.apache.iotdb.udf.api.UDTF;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.SocketTimeoutException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;

import static org.apache.iotdb.db.client.ConfigNodeClient.MSG_RECONNECTION_FAIL;

public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {

    private static final Logger LOGGER = LoggerFactory.getLogger(ClusterConfigTaskExecutor.class);

    private static final IClientManager<ConfigNodeRegionId, ConfigNodeClient>
            CONFIG_NODE_CLIENT_MANAGER =
            new IClientManager.Factory<ConfigNodeRegionId, ConfigNodeClient>()
                    .createClientManager(new DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());

    private static final IClientManager<ConfigNodeRegionId, ConfigNodeClient>
            CLUSTER_DELETION_CONFIG_NODE_CLIENT_MANAGER =
            new IClientManager.Factory<ConfigNodeRegionId, ConfigNodeClient>()
                    .createClientManager(
                            new DataNodeClientPoolFactory.ClusterDeletionConfigNodeClientPoolFactory());

    private static final class ClusterConfigTaskExecutorHolder {
        private static final ClusterConfigTaskExecutor INSTANCE = new ClusterConfigTaskExecutor();

        private ClusterConfigTaskExecutorHolder() {
        }
    }

    public static ClusterConfigTaskExecutor getInstance() {
        return ClusterConfigTaskExecutor.ClusterConfigTaskExecutorHolder.INSTANCE;
    }

    @Override
    public SettableFuture<ConfigTaskResult> setStorageGroup(
            SetStorageGroupStatement setStorageGroupStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        // Construct request using statement
        TStorageGroupSchema storageGroupSchema =
                SetStorageGroupTask.constructStorageGroupSchema(setStorageGroupStatement);
        TSetStorageGroupReq req = new TSetStorageGroupReq(storageGroupSchema);
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            // Send request to some API server
            TSStatus tsStatus = configNodeClient.setStorageGroup(req);
            // Get response or throw exception
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute create database {} in config node, status is {}.",
                        setStorageGroupStatement.getStorageGroupPath(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showStorageGroup(
            ShowStorageGroupStatement showStorageGroupStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        // Construct request using statement
        List<String> storageGroupPathPattern =
                Arrays.asList(showStorageGroupStatement.getPathPattern().getNodes());
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            // Send request to some API server
            TShowStorageGroupResp resp = client.showStorageGroup(storageGroupPathPattern);
            // build TSBlock
            ShowStorageGroupTask.buildTSBlock(resp.getStorageGroupInfoMap(), future);
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> countStorageGroup(
            CountStorageGroupStatement countStorageGroupStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        int storageGroupNum;
        List<String> storageGroupPathPattern =
                Arrays.asList(countStorageGroupStatement.getPathPattern().getNodes());
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TCountStorageGroupResp resp = client.countMatchedStorageGroups(storageGroupPathPattern);
            storageGroupNum = resp.getCount();
            // build TSBlock
            CountStorageGroupTask.buildTSBlock(storageGroupNum, future);
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> deleteStorageGroup(
            DeleteStorageGroupStatement deleteStorageGroupStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TDeleteStorageGroupsReq req =
                new TDeleteStorageGroupsReq(deleteStorageGroupStatement.getPrefixPath());
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TSStatus tsStatus = client.deleteStorageGroups(req);
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute delete database {} in config node, status is {}.",
                        deleteStorageGroupStatement.getPrefixPath(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> createFunction(
            CreateFunctionStatement createFunctionStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        String udfName = createFunctionStatement.getUdfName();
        String className = createFunctionStatement.getClassName();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TCreateFunctionReq tCreateFunctionReq = new TCreateFunctionReq(udfName, className, false);
            String libRoot = UDFExecutableManager.getInstance().getLibRoot();
            String jarFileName;
            ByteBuffer jarFile;
            String jarMd5;
            if (createFunctionStatement.isUsingURI()) {
                String uriString = createFunctionStatement.getUriString();
                if (uriString == null || uriString.isEmpty()) {
                    future.setException(
                            new IoTDBException(
                                    "URI is empty, please specify the URI.",
                                    TSStatusCode.TRIGGER_DOWNLOAD_ERROR.getStatusCode()));
                    return future;
                }
                jarFileName = new File(uriString).getName();
                try {
                    URI uri = new URI(uriString);
                    if (uri.getScheme() == null) {
                        future.setException(
                                new IoTDBException(
                                        "The scheme of URI is not set, please specify the scheme of URI.",
                                        TSStatusCode.TRIGGER_DOWNLOAD_ERROR.getStatusCode()));
                        return future;
                    }
                    if (!uri.getScheme().equals("file")) {
                        // Download executable
                        ExecutableResource resource =
                                UDFExecutableManager.getInstance().request(Collections.singletonList(uriString));
                        String jarFilePathUnderTempDir =
                                UDFExecutableManager.getInstance()
                                        .getDirStringUnderTempRootByRequestId(resource.getRequestId())
                                        + File.separator
                                        + jarFileName;
                        // libRoot should be the path of the specified jar
                        libRoot = jarFilePathUnderTempDir;
                        jarFile = ExecutableManager.transferToBytebuffer(jarFilePathUnderTempDir);
                        jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(jarFilePathUnderTempDir)));
                    } else {
                        // libRoot should be the path of the specified jar
                        libRoot = new File(new URI(uriString)).getAbsolutePath();
                        // If jarPath is a file path on datanode, we transfer it to ByteBuffer and send it to
                        // ConfigNode.
                        jarFile = ExecutableManager.transferToBytebuffer(libRoot);
                        // Set md5 of the jar file
                        jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(libRoot)));
                    }
                } catch (IOException | URISyntaxException e) {
                    LOGGER.warn(
                            "Failed to get executable for UDF({}) using URI: {}, the cause is: {}",
                            createFunctionStatement.getUdfName(),
                            createFunctionStatement.getUriString(),
                            e);
                    future.setException(
                            new IoTDBException(
                                    "Failed to get executable for UDF '"
                                            + createFunctionStatement.getUdfName()
                                            + "', please check the URI.",
                                    TSStatusCode.TRIGGER_DOWNLOAD_ERROR.getStatusCode()));
                    return future;
                }
                // modify req
                tCreateFunctionReq.setJarFile(jarFile);
                tCreateFunctionReq.setJarMD5(jarMd5);
                tCreateFunctionReq.setIsUsingURI(true);
                tCreateFunctionReq.setJarName(
                        String.format(
                                "%s-%s.%s",
                                jarFileName.substring(0, jarFileName.lastIndexOf(".")),
                                jarMd5,
                                jarFileName.substring(jarFileName.lastIndexOf(".") + 1)));
            }

            // try to create instance, this request will fail if creation is not successful
            try (UDFClassLoader classLoader = new UDFClassLoader(libRoot)) {
                // ensure that jar file contains the class and the class is a UDF
                Class<?> clazz = Class.forName(createFunctionStatement.getClassName(), true, classLoader);
                UDTF udtf = (UDTF) clazz.getDeclaredConstructor().newInstance();
            } catch (ClassNotFoundException
                    | NoSuchMethodException
                    | InstantiationException
                    | IllegalAccessException
                    | InvocationTargetException
                    | ClassCastException e) {
                LOGGER.warn(
                        "Failed to create function when try to create UDF({}) instance first, the cause is: {}",
                        createFunctionStatement.getUdfName(),
                        e);
                future.setException(
                        new IoTDBException(
                                "Failed to load class '"
                                        + createFunctionStatement.getClassName()
                                        + "', because it's not found in jar file: "
                                        + createFunctionStatement.getUriString(),
                                TSStatusCode.UDF_LOAD_CLASS_ERROR.getStatusCode()));
                return future;
            }

            final TSStatus executionStatus = client.createFunction(tCreateFunctionReq);
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
                LOGGER.warn(
                        "Failed to create function {}({}) because {}",
                        udfName,
                        className,
                        executionStatus.getMessage());
                future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | IOException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> dropFunction(String udfName) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            final TSStatus executionStatus = client.dropFunction(new TDropFunctionReq(udfName));

            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
                LOGGER.warn("[{}] Failed to drop function {}.", executionStatus, udfName);
                future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showFunctions() {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TGetUDFTableResp getUDFTableResp = client.getUDFTable();
            if (getUDFTableResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(
                        new IoTDBException(
                                getUDFTableResp.getStatus().message, getUDFTableResp.getStatus().code));
                return future;
            }
            // convert triggerTable and buildTsBlock
            ShowFunctionsTask.buildTsBlock(getUDFTableResp.getAllUDFInformation(), future);
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }

        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> createTrigger(
            CreateTriggerStatement createTriggerStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {

            TCreateTriggerReq tCreateTriggerReq =
                    new TCreateTriggerReq(
                            createTriggerStatement.getTriggerName(),
                            createTriggerStatement.getClassName(),
                            createTriggerStatement.getTriggerEvent().getId(),
                            createTriggerStatement.getTriggerType().getId(),
                            createTriggerStatement.getPathPattern().serialize(),
                            createTriggerStatement.getAttributes(),
                            FailureStrategy.OPTIMISTIC.getId(),
                            createTriggerStatement.isUsingURI()); // set default strategy

            String libRoot = TriggerExecutableManager.getInstance().getLibRoot();
            String jarFileName;
            ByteBuffer jarFile;
            String jarMd5;
            if (createTriggerStatement.isUsingURI()) {
                String uriString = createTriggerStatement.getUriString();
                if (uriString == null || uriString.isEmpty()) {
                    future.setException(
                            new IoTDBException(
                                    "URI is empty, please specify the URI.",
                                    TSStatusCode.UDF_DOWNLOAD_ERROR.getStatusCode()));
                    return future;
                }
                jarFileName = new File(uriString).getName();
                try {
                    URI uri = new URI(uriString);
                    if (uri.getScheme() == null) {
                        future.setException(
                                new IoTDBException(
                                        "The scheme of URI is not set, please specify the scheme of URI.",
                                        TSStatusCode.TRIGGER_DOWNLOAD_ERROR.getStatusCode()));
                        return future;
                    }
                    if (!uri.getScheme().equals("file")) {
                        // download executable
                        ExecutableResource resource =
                                TriggerExecutableManager.getInstance()
                                        .request(Collections.singletonList(uriString));
                        String jarFilePathUnderTempDir =
                                TriggerExecutableManager.getInstance()
                                        .getDirStringUnderTempRootByRequestId(resource.getRequestId())
                                        + File.separator
                                        + jarFileName;
                        // libRoot should be the path of the specified jar
                        libRoot = jarFilePathUnderTempDir;
                        jarFile = ExecutableManager.transferToBytebuffer(jarFilePathUnderTempDir);
                        jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(jarFilePathUnderTempDir)));

                    } else {
                        // libRoot should be the path of the specified jar
                        libRoot = new File(new URI(uriString)).getAbsolutePath();
                        // If jarPath is a file path on datanode, we transfer it to ByteBuffer and send it to
                        // ConfigNode.
                        jarFile = ExecutableManager.transferToBytebuffer(libRoot);
                        // set md5 of the jar file
                        jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(libRoot)));
                    }
                } catch (IOException | URISyntaxException e) {
                    LOGGER.warn(
                            "Failed to get executable for Trigger({}) using URI: {}, the cause is: {}",
                            createTriggerStatement.getTriggerName(),
                            createTriggerStatement.getUriString(),
                            e);
                    future.setException(
                            new IoTDBException(
                                    "Failed to get executable for Trigger '"
                                            + createTriggerStatement.getUriString()
                                            + "', please check the URI.",
                                    TSStatusCode.TRIGGER_DOWNLOAD_ERROR.getStatusCode()));
                    return future;
                }
                // modify req
                tCreateTriggerReq.setJarFile(jarFile);
                tCreateTriggerReq.setJarMD5(jarMd5);
                tCreateTriggerReq.setIsUsingURI(true);
                tCreateTriggerReq.setJarName(
                        String.format(
                                "%s-%s.%s",
                                jarFileName.substring(0, jarFileName.lastIndexOf(".")),
                                jarMd5,
                                jarFileName.substring(jarFileName.lastIndexOf(".") + 1)));
            }

            // try to create instance, this request will fail if creation is not successful
            try (TriggerClassLoader classLoader = new TriggerClassLoader(libRoot)) {
                Class<?> triggerClass =
                        Class.forName(createTriggerStatement.getClassName(), true, classLoader);
                Trigger trigger = (Trigger) triggerClass.getDeclaredConstructor().newInstance();
                tCreateTriggerReq.setFailureStrategy(trigger.getFailureStrategy().getId());
            } catch (ClassNotFoundException
                    | NoSuchMethodException
                    | InstantiationException
                    | IllegalAccessException
                    | InvocationTargetException
                    | ClassCastException e) {
                LOGGER.warn(
                        "Failed to create trigger when try to create trigger({}) instance first, the cause is: {}",
                        createTriggerStatement.getTriggerName(),
                        e);
                future.setException(
                        new IoTDBException(
                                "Failed to load class '"
                                        + createTriggerStatement.getClassName()
                                        + "', because it's not found in jar file: "
                                        + createTriggerStatement.getUriString(),
                                TSStatusCode.TRIGGER_LOAD_CLASS_ERROR.getStatusCode()));
                return future;
            }

            final TSStatus executionStatus = client.createTrigger(tCreateTriggerReq);

            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
                LOGGER.warn(
                        "[{}] Failed to create trigger {}. TSStatus is {}",
                        executionStatus,
                        createTriggerStatement.getTriggerName(),
                        executionStatus.message);
                future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException | IOException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> dropTrigger(String triggerName) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            final TSStatus executionStatus = client.dropTrigger(new TDropTriggerReq(triggerName));
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
                LOGGER.warn("[{}] Failed to drop trigger {}.", executionStatus, triggerName);
                future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showTriggers() {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TGetTriggerTableResp getTriggerTableResp = client.getTriggerTable();
            if (getTriggerTableResp.getStatus().getCode()
                    != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(
                        new IoTDBException(
                                getTriggerTableResp.getStatus().message, getTriggerTableResp.getStatus().code));
                return future;
            }
            // convert triggerTable and buildTsBlock
            ShowTriggersTask.buildTsBlock(getTriggerTableResp.getAllTriggerInformation(), future);
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }

        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> setTTL(SetTTLStatement setTTLStatement, String taskName) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        List<String> storageGroupPathPattern =
                Arrays.asList(setTTLStatement.getStorageGroupPath().getNodes());
        TSetTTLReq setTTLReq = new TSetTTLReq(storageGroupPathPattern, setTTLStatement.getTTL());
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            // Send request to some API server
            TSStatus tsStatus = configNodeClient.setTTL(setTTLReq);
            // Get response or throw exception
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute {} {} in config node, status is {}.",
                        taskName,
                        setTTLStatement.getStorageGroupPath(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> merge(boolean onCluster) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TSStatus tsStatus = new TSStatus();
        if (onCluster) {
            try (ConfigNodeClient client =
                         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
                // Send request to some API server
                tsStatus = client.merge();
            } catch (ClientManagerException | TException e) {
                future.setException(e);
            }
        } else {
            tsStatus = LocalConfigNode.getInstance().executeMergeOperation();
        }
        if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
            future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
        } else {
            future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> flush(TFlushReq tFlushReq, boolean onCluster) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TSStatus tsStatus = new TSStatus();
        if (onCluster) {
            try (ConfigNodeClient client =
                         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
                // Send request to some API server
                tsStatus = client.flush(tFlushReq);
            } catch (ClientManagerException | TException e) {
                future.setException(e);
            }
        } else {
            tsStatus = LocalConfigNode.getInstance().executeFlushOperation(tFlushReq);
        }
        if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
            future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
        } else {
            future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> clearCache(boolean onCluster) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TSStatus tsStatus = new TSStatus();
        if (onCluster) {
            try (ConfigNodeClient client =
                         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
                // Send request to some API server
                tsStatus = client.clearCache();
            } catch (ClientManagerException | TException e) {
                future.setException(e);
            }
        } else {
            tsStatus = LocalConfigNode.getInstance().executeClearCacheOperation();
        }
        if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
            future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
        } else {
            future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> loadConfiguration(boolean onCluster) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TSStatus tsStatus = new TSStatus();
        if (onCluster) {
            try (ConfigNodeClient client =
                         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
                // Send request to some API server
                tsStatus = client.loadConfiguration();
            } catch (ClientManagerException | TException e) {
                future.setException(e);
            }
        } else {
            tsStatus = LocalConfigNode.getInstance().executeLoadConfigurationOperation();
        }
        if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
            future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
        } else {
            future.setException(new StatementExecutionException(tsStatus));
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> setSystemStatus(boolean onCluster, NodeStatus status) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TSStatus tsStatus = new TSStatus();
        if (onCluster) {
            try (ConfigNodeClient client =
                         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
                // Send request to some API server
                tsStatus = client.setSystemStatus(status.getStatus());
            } catch (ClientManagerException | TException e) {
                future.setException(e);
            }
        } else {
            tsStatus = LocalConfigNode.getInstance().executeSetSystemStatus(status);
        }
        if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
            future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
        } else {
            future.setException(new StatementExecutionException(tsStatus));
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showCluster(ShowClusterStatement showClusterStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TShowClusterResp showClusterResp = new TShowClusterResp();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            showClusterResp = client.showCluster();
        } catch (ClientManagerException | TException e) {
            if (showClusterResp.getConfigNodeList() == null) {
                future.setException(new TException(MSG_RECONNECTION_FAIL));
            } else {
                future.setException(e);
            }
            return future;
        }
        // build TSBlock
        if (showClusterStatement.isDetails()) {
            ShowClusterDetailsTask.buildTSBlock(showClusterResp, future);
        } else {
            ShowClusterTask.buildTSBlock(showClusterResp, future);
        }

        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showClusterParameters() {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TShowClusterParametersResp showClusterParametersResp = new TShowClusterParametersResp();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            showClusterParametersResp = client.showClusterParameters();
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }

        // build TSBlock
        ShowClusterParametersTask.buildTSBlock(showClusterParametersResp, future);

        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showTTL(ShowTTLStatement showTTLStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        List<PartialPath> storageGroupPaths = showTTLStatement.getPaths();
        Map<String, Long> storageGroupToTTL = new HashMap<>();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            if (showTTLStatement.isAll()) {
                List<String> allStorageGroupPathPattern = Arrays.asList("root", "**");
                TStorageGroupSchemaResp resp =
                        client.getMatchedStorageGroupSchemas(allStorageGroupPathPattern);
                for (Map.Entry<String, TStorageGroupSchema> entry :
                        resp.getStorageGroupSchemaMap().entrySet()) {
                    storageGroupToTTL.put(entry.getKey(), entry.getValue().getTTL());
                }
            } else {
                for (PartialPath storageGroupPath : storageGroupPaths) {
                    List<String> storageGroupPathPattern = Arrays.asList(storageGroupPath.getNodes());
                    TStorageGroupSchemaResp resp =
                            client.getMatchedStorageGroupSchemas(storageGroupPathPattern);
                    for (Map.Entry<String, TStorageGroupSchema> entry :
                            resp.getStorageGroupSchemaMap().entrySet()) {
                        if (!storageGroupToTTL.containsKey(entry.getKey())) {
                            storageGroupToTTL.put(entry.getKey(), entry.getValue().getTTL());
                        }
                    }
                }
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        // build TSBlock
        ShowTTLTask.buildTSBlock(storageGroupToTTL, future);
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showRegion(ShowRegionStatement showRegionStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TShowRegionResp showRegionResp = new TShowRegionResp();
        TShowRegionReq showRegionReq = new TShowRegionReq();
        showRegionReq.setConsensusGroupType(showRegionStatement.getRegionType());
        if (showRegionStatement.getStorageGroups() == null) {
            showRegionReq.setStorageGroups(null);
        } else {
            showRegionReq.setStorageGroups(
                    showRegionStatement.getStorageGroups().stream()
                            .map(PartialPath::getFullPath)
                            .collect(Collectors.toList()));
        }
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            showRegionResp = client.showRegion(showRegionReq);
            if (showRegionResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(
                        new IoTDBException(
                                showRegionResp.getStatus().message, showRegionResp.getStatus().code));
                return future;
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        // build TSBlock
        ShowRegionTask.buildTSBlock(showRegionResp, future);
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showDataNodes(
            ShowDataNodesStatement showDataNodesStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TShowDataNodesResp showDataNodesResp = new TShowDataNodesResp();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            showDataNodesResp = client.showDataNodes();
            if (showDataNodesResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(
                        new IoTDBException(
                                showDataNodesResp.getStatus().message, showDataNodesResp.getStatus().code));
                return future;
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        // build TSBlock
        ShowDataNodesTask.buildTSBlock(showDataNodesResp, future);
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showConfigNodes() {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TShowConfigNodesResp showConfigNodesResp = new TShowConfigNodesResp();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            showConfigNodesResp = client.showConfigNodes();
            if (showConfigNodesResp.getStatus().getCode()
                    != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(
                        new IoTDBException(
                                showConfigNodesResp.getStatus().message, showConfigNodesResp.getStatus().code));
                return future;
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        // build TSBlock
        ShowConfigNodesTask.buildTSBlock(showConfigNodesResp, future);
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> createSchemaTemplate(
            CreateSchemaTemplateStatement createSchemaTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        // Construct request using statement
        try {
            // Send request to some API server
            TSStatus tsStatus =
                    ClusterTemplateManager.getInstance().createSchemaTemplate(createSchemaTemplateStatement);
            // Get response or throw exception
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute create schema template {} in config node, status is {}.",
                        createSchemaTemplateStatement.getName(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e.getCause());
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showSchemaTemplate(
            ShowSchemaTemplateStatement showSchemaTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try {
            // Send request to some API server
            List<Template> templateList = ClusterTemplateManager.getInstance().getAllTemplates();
            // build TSBlock
            ShowSchemaTemplateTask.buildTSBlock(templateList, future);
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showNodesInSchemaTemplate(
            ShowNodesInSchemaTemplateStatement showNodesInSchemaTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        String req = showNodesInSchemaTemplateStatement.getTemplateName();
        try {
            // Send request to some API server
            Template template = ClusterTemplateManager.getInstance().getTemplate(req);
            // Build TSBlock
            ShowNodesInSchemaTemplateTask.buildTSBlock(template, future);
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> setSchemaTemplate(
            SetSchemaTemplateStatement setSchemaTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        String templateName = setSchemaTemplateStatement.getTemplateName();
        PartialPath path = setSchemaTemplateStatement.getPath();
        try {
            // Send request to some API server
            ClusterTemplateManager.getInstance().setSchemaTemplate(templateName, path);
            // build TSBlock
            future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
        } catch (Exception e) {
            future.setException(e.getCause());
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showPathSetTemplate(
            ShowPathSetTemplateStatement showPathSetTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        String templateName = showPathSetTemplateStatement.getTemplateName();
        try {
            // Send request to some API server
            List<PartialPath> listPath =
                    ClusterTemplateManager.getInstance().getPathsSetTemplate(templateName);
            // Build TSBlock
            ShowPathSetTemplateTask.buildTSBlock(listPath, future);
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> deactivateSchemaTemplate(
            String queryId, DeactivateTemplateStatement deactivateTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TDeactivateSchemaTemplateReq req = new TDeactivateSchemaTemplateReq();
        req.setQueryId(queryId);
        req.setTemplateName(deactivateTemplateStatement.getTemplateName());
        req.setPathPatternTree(
                serializePatternListToByteBuffer(deactivateTemplateStatement.getPathPatternList()));
        try (ConfigNodeClient client =
                     CLUSTER_DELETION_CONFIG_NODE_CLIENT_MANAGER.borrowClient(
                             ConfigNodeInfo.configNodeRegionId)) {
            TSStatus tsStatus;
            do {
                try {
                    tsStatus = client.deactivateSchemaTemplate(req);
                } catch (TTransportException e) {
                    if (e.getType() == TTransportException.TIMED_OUT
                            || e.getCause() instanceof SocketTimeoutException) {
                        // Time out mainly caused by slow execution, just wait
                        tsStatus = RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK);
                    } else {
                        throw e;
                    }
                }
                // Keep waiting until task ends
            } while (TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode() == tsStatus.getCode());

            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute deactivate schema template {} from {} in config node, status is {}.",
                        deactivateTemplateStatement.getTemplateName(),
                        deactivateTemplateStatement.getPathPatternList(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> dropSchemaTemplate(
            DropSchemaTemplateStatement dropSchemaTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            // Send request to some API server
            TSStatus tsStatus =
                    configNodeClient.dropSchemaTemplate(dropSchemaTemplateStatement.getTemplateName());
            // Get response or throw exception
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute drop schema template {} in config node, status is {}.",
                        dropSchemaTemplateStatement.getTemplateName(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    private ByteBuffer serializePatternListToByteBuffer(List<PartialPath> patternList) {
        PathPatternTree patternTree = new PathPatternTree();
        for (PartialPath pathPattern : patternList) {
            patternTree.appendPathPattern(pathPattern);
        }
        patternTree.constructTree();
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
        try {
            patternTree.serialize(dataOutputStream);
        } catch (IOException ignored) {
            // memory operation, won't happen
        }
        return ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
    }

    @Override
    public SettableFuture<ConfigTaskResult> unsetSchemaTemplate(
            String queryId, UnsetSchemaTemplateStatement unsetSchemaTemplateStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TUnsetSchemaTemplateReq req = new TUnsetSchemaTemplateReq();
        req.setQueryId(queryId);
        req.setTemplateName(unsetSchemaTemplateStatement.getTemplateName());
        req.setPath(unsetSchemaTemplateStatement.getPath().getFullPath());
        try (ConfigNodeClient client =
                     CLUSTER_DELETION_CONFIG_NODE_CLIENT_MANAGER.borrowClient(
                             ConfigNodeInfo.configNodeRegionId)) {
            TSStatus tsStatus;
            do {
                try {
                    tsStatus = client.unsetSchemaTemplate(req);
                } catch (TTransportException e) {
                    if (e.getType() == TTransportException.TIMED_OUT
                            || e.getCause() instanceof SocketTimeoutException) {
                        // time out mainly caused by slow execution, wait until
                        tsStatus = RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK);
                    } else {
                        throw e;
                    }
                }
                // keep waiting until task ends
            } while (TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode() == tsStatus.getCode());

            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute unset schema template {} from {} in config node, status is {}.",
                        unsetSchemaTemplateStatement.getTemplateName(),
                        unsetSchemaTemplateStatement.getPath(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> createPipeSink(
            CreatePipeSinkStatement createPipeSinkStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TPipeSinkInfo pipeSinkInfo = new TPipeSinkInfo();
            pipeSinkInfo.setPipeSinkName(createPipeSinkStatement.getPipeSinkName());
            pipeSinkInfo.setPipeSinkType(createPipeSinkStatement.getPipeSinkType());
            pipeSinkInfo.setAttributes(createPipeSinkStatement.getAttributes());
            TSStatus tsStatus = configNodeClient.createPipeSink(pipeSinkInfo);
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to create PIPESINK {} with type {} in config node, status is {}.",
                        createPipeSinkStatement.getPipeSinkName(),
                        createPipeSinkStatement.getPipeSinkType(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> dropPipeSink(
            DropPipeSinkStatement dropPipeSinkStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TDropPipeSinkReq req = new TDropPipeSinkReq();
            req.setPipeSinkName(dropPipeSinkStatement.getPipeSinkName());
            TSStatus tsStatus = configNodeClient.dropPipeSink(req);
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to drop PIPESINK {} in config node, status is {}.",
                        dropPipeSinkStatement.getPipeSinkName(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showPipeSink(
            ShowPipeSinkStatement showPipeSinkStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TGetPipeSinkReq tGetPipeSinkReq = new TGetPipeSinkReq();
            if (!StringUtils.isEmpty(showPipeSinkStatement.getPipeSinkName())) {
                tGetPipeSinkReq.setPipeSinkName(showPipeSinkStatement.getPipeSinkName());
            }
            TGetPipeSinkResp resp = configNodeClient.getPipeSink(tGetPipeSinkReq);
            ShowPipeSinkTask.buildTSBlockByTPipeSinkInfo(resp.getPipeSinkInfoList(), future);
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> createPipe(CreatePipeStatement createPipeStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TCreatePipeReq req =
                    new TCreatePipeReq()
                            .setPipeName(createPipeStatement.getPipeName())
                            .setPipeSinkName(createPipeStatement.getPipeSinkName())
                            .setStartTime(createPipeStatement.getStartTime())
                            .setAttributes(createPipeStatement.getPipeAttributes());
            TSStatus tsStatus = configNodeClient.createPipe(req);
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to create PIPE {} in config node, status is {}.",
                        createPipeStatement.getPipeName(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> startPipe(StartPipeStatement startPipeStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TSStatus tsStatus = configNodeClient.startPipe(startPipeStatement.getPipeName());
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to start PIPE {}, status is {}.", startPipeStatement.getPipeName(), tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> dropPipe(DropPipeStatement dropPipeStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TSStatus tsStatus = configNodeClient.dropPipe(dropPipeStatement.getPipeName());
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to drop PIPE {}, status is {}.", dropPipeStatement.getPipeName(), tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> stopPipe(StopPipeStatement stopPipeStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TSStatus tsStatus = configNodeClient.stopPipe(stopPipeStatement.getPipeName());
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to stop PIPE {}, status is {}.", stopPipeStatement.getPipeName(), tsStatus);
                future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showPipe(ShowPipeStatement showPipeStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TShowPipeReq tShowPipeReq = new TShowPipeReq();
            if (!StringUtils.isEmpty(showPipeStatement.getPipeName())) {
                tShowPipeReq.setPipeName(showPipeStatement.getPipeName());
            }
            TShowPipeResp resp = configNodeClient.showPipe(tShowPipeReq);
            List<TShowPipeInfo> tShowPipeInfoList =
                    SyncService.getInstance().showPipeForReceiver(showPipeStatement.getPipeName());
            tShowPipeInfoList.addAll(resp.getPipeInfoList());
            ShowPipeTask.buildTSBlock(tShowPipeInfoList, future);
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> deleteTimeSeries(
            String queryId, DeleteTimeSeriesStatement deleteTimeSeriesStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TDeleteTimeSeriesReq req =
                new TDeleteTimeSeriesReq(
                        queryId,
                        serializePatternListToByteBuffer(deleteTimeSeriesStatement.getPathPatternList()));
        try (ConfigNodeClient client =
                     CLUSTER_DELETION_CONFIG_NODE_CLIENT_MANAGER.borrowClient(
                             ConfigNodeInfo.configNodeRegionId)) {
            TSStatus tsStatus;
            do {
                try {
                    tsStatus = client.deleteTimeSeries(req);
                } catch (TTransportException e) {
                    if (e.getType() == TTransportException.TIMED_OUT
                            || e.getCause() instanceof SocketTimeoutException) {
                        // time out mainly caused by slow execution, wait until
                        tsStatus = RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK);
                    } else {
                        throw e;
                    }
                }
                // keep waiting until task ends
            } while (TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode() == tsStatus.getCode());

            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
                LOGGER.warn(
                        "Failed to execute delete timeseries {} in config node, status is {}.",
                        deleteTimeSeriesStatement.getPathPatternList(),
                        tsStatus);
                future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> getRegionId(GetRegionIdStatement getRegionIdStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TGetRegionIdResp resp = new TGetRegionIdResp();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TGetRegionIdReq tGetRegionIdReq =
                    new TGetRegionIdReq(
                            getRegionIdStatement.getStorageGroup(), getRegionIdStatement.getPartitionType());
            if (getRegionIdStatement.getSeriesSlotId() != null) {
                tGetRegionIdReq.setSeriesSlotId(getRegionIdStatement.getSeriesSlotId());
            } else {
                tGetRegionIdReq.setDeviceId(getRegionIdStatement.getDeviceId());
            }
            if (getRegionIdStatement.getTimeSlotId() != null) {
                tGetRegionIdReq.setTimeSlotId(getRegionIdStatement.getTimeSlotId());
            } else if (getRegionIdStatement.getTimeStamp() != -1) {
                tGetRegionIdReq.setTimeStamp(getRegionIdStatement.getTimeStamp());
            }
            resp = configNodeClient.getRegionId(tGetRegionIdReq);
            if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(new IoTDBException(resp.getStatus().message, resp.getStatus().code));
                return future;
            }
        } catch (Exception e) {
            future.setException(e);
        }
        GetRegionIdTask.buildTSBlock(resp, future);
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> getSeriesSlotList(
            GetSeriesSlotListStatement getSeriesSlotListStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TGetSeriesSlotListResp resp = new TGetSeriesSlotListResp();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TGetSeriesSlotListReq tGetSeriesSlotListReq =
                    new TGetSeriesSlotListReq(getSeriesSlotListStatement.getStorageGroup());
            if (getSeriesSlotListStatement.getPartitionType() != null) {
                tGetSeriesSlotListReq.setType(getSeriesSlotListStatement.getPartitionType());
            }
            resp = configNodeClient.getSeriesSlotList(tGetSeriesSlotListReq);
            if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(new IoTDBException(resp.getStatus().message, resp.getStatus().code));
                return future;
            }
        } catch (Exception e) {
            future.setException(e);
        }
        GetSeriesSlotListTask.buildTSBlock(resp, future);
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> getTimeSlotList(
            GetTimeSlotListStatement getTimeSlotListStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        TGetTimeSlotListResp resp = new TGetTimeSlotListResp();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TGetTimeSlotListReq tGetTimeSlotListReq =
                    new TGetTimeSlotListReq(
                            getTimeSlotListStatement.getStorageGroup(),
                            getTimeSlotListStatement.getSeriesSlotId());
            if (getTimeSlotListStatement.getStartTime() != -1) {
                tGetTimeSlotListReq.setStartTime(getTimeSlotListStatement.getStartTime());
            }
            if (getTimeSlotListStatement.getEndTime() != -1) {
                tGetTimeSlotListReq.setEndTime(getTimeSlotListStatement.getEndTime());
            }
            resp = configNodeClient.getTimeSlotList(tGetTimeSlotListReq);
            if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(new IoTDBException(resp.getStatus().message, resp.getStatus().code));
                return future;
            }
        } catch (Exception e) {
            future.setException(e);
        }
        GetTimeSlotListTask.buildTSBlock(resp, future);
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> migrateRegion(
            MigrateRegionStatement migrateRegionStatement) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient configNodeClient =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TMigrateRegionReq tMigrateRegionReq =
                    new TMigrateRegionReq(
                            migrateRegionStatement.getRegionId(),
                            migrateRegionStatement.getFromId(),
                            migrateRegionStatement.getToId());
            TSStatus status = configNodeClient.migrateRegion(tMigrateRegionReq);
            if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(new IoTDBException(status.message, status.code));
                return future;
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (Exception e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> createContinuousQuery(
            CreateContinuousQueryStatement createContinuousQueryStatement, String sql, String username) {
        createContinuousQueryStatement.semanticCheck();

        String queryBody = createContinuousQueryStatement.getQueryBody();
        // TODO Do not modify Statement in Analyzer
        Analyzer.validate(createContinuousQueryStatement.getQueryBodyStatement());

        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TCreateCQReq tCreateCQReq =
                    new TCreateCQReq(
                            createContinuousQueryStatement.getCqId(),
                            createContinuousQueryStatement.getEveryInterval(),
                            createContinuousQueryStatement.getBoundaryTime(),
                            createContinuousQueryStatement.getStartTimeOffset(),
                            createContinuousQueryStatement.getEndTimeOffset(),
                            createContinuousQueryStatement.getTimeoutPolicy().getType(),
                            queryBody,
                            sql,
                            createContinuousQueryStatement.getZoneId(),
                            username);
            final TSStatus executionStatus = client.createCQ(tCreateCQReq);
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
                LOGGER.warn(
                        "[{}] Failed to create continuous query {}. TSStatus is {}",
                        executionStatus,
                        createContinuousQueryStatement.getCqId(),
                        executionStatus.message);
                future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> dropContinuousQuery(String cqId) {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            final TSStatus executionStatus = client.dropCQ(new TDropCQReq(cqId));
            if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
                LOGGER.warn("[{}] Failed to drop continuous query {}.", executionStatus, cqId);
                future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
            } else {
                future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
            }
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }
        return future;
    }

    @Override
    public SettableFuture<ConfigTaskResult> showContinuousQueries() {
        SettableFuture<ConfigTaskResult> future = SettableFuture.create();
        try (ConfigNodeClient client =
                     CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
            TShowCQResp showCQResp = client.showCQ();
            if (showCQResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                future.setException(
                        new IoTDBException(showCQResp.getStatus().message, showCQResp.getStatus().code));
                return future;
            }
            // convert cqList and buildTsBlock
            ShowContinuousQueriesTask.buildTsBlock(showCQResp.getCqList(), future);
        } catch (ClientManagerException | TException e) {
            future.setException(e);
        }

        return future;
    }
}
